package com.mldong.sse.impl;

import com.mldong.exception.ServiceException;
import com.mldong.sse.SseCache;
import com.mldong.sse.SseModel;
import com.mldong.sse.SseService;
import com.mldong.web.LoginUserHolder;
import lombok.RequiredArgsConstructor;
import org.jetbrains.annotations.NotNull;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

@Service
@RequiredArgsConstructor
public class SseServiceImpl implements SseService, MessageListener {
    private final Map<String,SseEmitter> emitters = new ConcurrentHashMap<>();
    private final SseCache sseCache;
    @PostConstruct
    public void init(){
        // 订阅特定的频道
        Objects.requireNonNull(sseCache.getRedisTemplate().getConnectionFactory()).getConnection().subscribe(this, sseCache.getCommonKeyPrefix().getBytes());
    }
    @Override
    public SseEmitter addEmitter(String token) {
        SseEmitter emitter = new SseEmitter();
        emitters.put(token,emitter);
        emitter.onCompletion(() -> emitters.remove(token));
        emitter.onTimeout(() -> emitters.remove(token));
        try {
            Long userId = LoginUserHolder.getUserId();
            emitter.send(SseModel.builder().userId(userId).type("init").msg("初始化连接成功").build());
        } catch (IOException e) {
            emitters.remove(token);
            ServiceException.throwBiz(99999999,"sse连接异常");
        }
        return emitter;
    }

    @Override
    public void broadcast(SseModel model) {
        sseCache.getRedisTemplate().convertAndSend(sseCache.getCommonKeyPrefix(), model);
    }

    @Override
    public void onMessage(@NotNull Message message, byte[] bytes) {
        emitters.forEach((key,emitter)->{
            try {
                emitter.send(message.getBody());
            } catch (IOException e) {
                emitters.remove(key);
            }
        });
    }
}
